package com.cn.nio协议;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.junit.Before;
import org.junit.Test;

import javax.jms.*;

/**
 * @author: dlr
 * @data: 2019/9/8 23:57
 * @desc:点对点消费
 */
public class pointTopointConsumer {
    private final String MQ_URL="nio://192.168.10.20:61618";
    //点对点对队列名
    private final String QUEUE_NAME="BASE_QUEUE";
    Connection connection=null;
    Session session=null;
    /*
     * @author: dlr
     * @desc:    获取连接
     * @date: 2019/8/11
     * @param
     * @return: void
     */
    @Before
    public void getConnection() throws Exception{
        //连接工厂
        ActiveMQConnectionFactory factory=new ActiveMQConnectionFactory(MQ_URL);
        //获取连接
        connection=factory.createConnection();
        //启动
        connection.start();
        //创建session
        //参数1:是否开启事务,参数2:签收模式
        session=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
    }
    /*
     * @author: dlr
     * @desc:   消费消息(receive方法接受)
     * @date: 2019/8/11
     * @param
     * @return: void
     */
    @Test
    public void consumerpTp()throws Exception{
        //创建消息目的地点对点队列
        Queue queue=session.createQueue(QUEUE_NAME);
        //创建消费者
        MessageConsumer consumer = session.createConsumer(queue);

        while(true){
            //等待4s没有消息了就关闭
            TextMessage receive =(TextMessage)consumer.receive(4000l);
            if(receive!=null){
                System.err.println(receive.getText());
            }
            else{
                break;
            }
        }
        consumer.close();
        session.close();
        connection.close();
    }
    /*
     * @author: dlr
     * @desc:     消费消息(listen监听)
     * @date: 2019/8/12
     * @param
     * @return:
     */
    @Test
    public void listenpTp()throws Exception {
        //创建消息目的地点对点队列
        Queue queue = session.createQueue(QUEUE_NAME);
        //创建消费者
        MessageConsumer consumer = session.createConsumer(queue);
        //监听队列
        consumer.setMessageListener(new MessageListener(){
            @Override
            public void onMessage(Message message) {
                TextMessage msg =(TextMessage)message;
                try{
                    if(msg!=null){
                        System.err.println(msg.getText());
                    }
                }
                catch (Exception e){
                    e.printStackTrace();
                }
            }
        });
        System.in.read();
        consumer.close();
        session.close();
        connection.close();
    }
}
